package xin.marcher.module.inventory.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import xin.marcher.module.common.constants.inventory.TairInventoryConstant;
import xin.marcher.module.common.core.JsonResult;
import xin.marcher.module.common.enums.inventory.InventoryDeductionTypeEnum;
import xin.marcher.module.common.enums.inventory.InventoryExceptionCode;
import xin.marcher.module.common.exception.BaseBizException;
import xin.marcher.module.common.tair.TairCache;
import xin.marcher.module.common.tair.TairLock;
import xin.marcher.module.common.utils.SnowflakeIdWorker;
import xin.marcher.module.inventory.cache.InventoryBucketCache;
import xin.marcher.module.inventory.converter.InventoryConverter;
import xin.marcher.module.inventory.domain.bo.*;
import xin.marcher.module.inventory.domain.entity.InventoryBucketConfigDO;
import xin.marcher.module.inventory.domain.request.InventoryQueryRequest;
import xin.marcher.module.inventory.domain.request.InventoryRequest;
import xin.marcher.module.inventory.domain.request.WarningInventoryRequest;
import xin.marcher.module.inventory.domain.response.InventoryResponseDTO;
import xin.marcher.module.inventory.generator.SegmentNoGen;
import xin.marcher.module.inventory.mq.producer.BucketCapacityProducer;
import xin.marcher.module.inventory.mq.producer.BucketRefreshProducer;
import xin.marcher.module.inventory.mq.producer.InventoryDetailProducer;
import xin.marcher.module.inventory.repository.InventoryRepository;
import xin.marcher.module.inventory.service.InventoryService;
import xin.marcher.module.inventory.util.InventorBucketUtil;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 库存扣减 业务实现类
 *
 *
 */
@Service
@Slf4j
public class InventoryServiceImpl implements InventoryService {

    @Resource
    private BucketCapacityProducer bucketCapacityProducer;
    @Resource
    private BucketRefreshProducer bucketRefreshProducer;

    @Resource
    private InventoryConverter inventoryConverter;

    @Resource
    private InventoryBucketCache inventoryBucketCache;

    @Resource
    private TairCache tairCache;

    @Resource
    private TairLock tairLock;

    @Resource
    private InventoryDetailProducer inventoryDetailProducer;

    @Resource
    private InventoryRepository inventoryRepository;

    @Resource
    private SegmentNoGen segmentNoGen;

    /**
     * 取消订单 回退商品库存
     *
     * @param request
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public JsonResult cancelProductStock(InventoryRequest request) {
        // 1. 验证入参
        checkDeductProductStock(request);
        String key = TairInventoryConstant.SELLER_INVENTORY_LOCK_PREFIX + request.getOrderId();
        String value = SnowflakeIdWorker.getCode();
        boolean lock = tairLock.tryLock(key, value);
        if (lock) {
            try {
                //2. 记录取消订单的库存申请记录
                InventoryDetail inventoryDetail = saveRefundInventoryDetail(request);
                //3.释放对应库存回到对应中心桶
                releaseInventory(inventoryDetail);
                return JsonResult.buildSuccess();
            } catch (Exception e) {
                e.printStackTrace();
                return JsonResult.buildError(e.getMessage());
            } finally {
                tairLock.unlock(key, value);
            }
        }
        return JsonResult.buildError("请求频繁，请稍后再试");
    }

    /**
     * 扣减商品库存
     *
     * @param request
     */
    @Override
    public JsonResult deductProductStock(InventoryRequest request) {
        // 1. 验证入参是否合法
        checkDeductProductStock(request);
        // 2. 构建扣减库存的上下文对象
        BucketContext bucketContext = buildDeductProductStock(request);

        try {
            // 3.获取是否已经有一条扣减明细记录
            String repeatDeductInfo = getRepeatDeductInfo(bucketContext);
            if (!StringUtils.isEmpty(repeatDeductInfo)) {
                return JsonResult.buildSuccess();
            }
            // 4.执行库存扣减
            deductInventory(bucketContext);
            // 5.写入明细，如果已重复写入失败，则回退库存
            writeInventoryDetail(bucketContext);
        } catch (Exception e) {
            log.error("库存扣减失败", e);
            return JsonResult.buildError(e.getMessage());
        } finally {
            // 6.检测当前返回的库存数量是否触发扩容的阈值(回源比例)，触发则异步通知扩容
            checkInventoryBackSource(bucketContext);
        }

        return JsonResult.buildSuccess();
    }

    /**
     * 库存预警消息通知
     *
     * @param warningInventoryRequest
     */
    @Override
    public void warningInventoryMessage(WarningInventoryRequest warningInventoryRequest) {
        // 此处暂不实现消息通知运营
        log.info("运营接收到库存预警消息通知:" + JSONObject.toJSON(warningInventoryRequest));
    }

    /**
     * 查询获取指定的卖家商品库存信息
     *
     * @param queryRequest
     * @return
     */
    @Override
    public List<InventoryResponseDTO> queryInventoryList(InventoryQueryRequest queryRequest) {
        //1.验证入参
        checkQueryInventoryList(queryRequest);
        //2.组装查询参数，查询卖家下商品的实际库存数量
        return queryInventoryCacheList(queryRequest);
    }

    /**
     * 查询返回卖家的商品实际库存信息
     *
     * @param queryRequest
     * @return
     */
    private List<InventoryResponseDTO> queryInventoryCacheList(InventoryQueryRequest queryRequest) {
        //1. 获取组装查询的缓存key
        List<String> skuIdList = queryRequest.getSkuIdList();
        List<InventoryResponseDTO> inventoryResponseDTOList = new ArrayList<>();
        for (String skuId : skuIdList) {
            // 卖家下的任意一个商品缓存信息
            String cacheKey = queryRequest.getSellerId() + skuId;
            // 缓存在本地的分桶信息
            BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(cacheKey);
            if (Objects.isNull(bucketLocalCache)) {
                continue;
            }
            // 组装当前这个商家对应的卖家商品分桶批量查询
            List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
            if (CollectionUtils.isEmpty(availableList)) {
                availableList = new ArrayList<>();
            }
            availableList.addAll(bucketLocalCache.getUndistributedList());
            List<String> bucketList = availableList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
            // 获取中心桶的库存
            String key = TairInventoryConstant.SELLER_INVENTORY_PREFIX + cacheKey;
            bucketList.add(key);
            // 获取对应的缓存库存
            List<String> bucketNumList = tairCache.mget(bucketList);
            Integer inventoryNum = 0;
            for (String bucketNum : bucketNumList) {
                if (!Objects.isNull(bucketNum)) {
                    inventoryNum = inventoryNum + Integer.valueOf(bucketNum);
                }
            }
            // 汇总商品库存构建模型
            InventoryResponseDTO inventoryResponseDTO = new InventoryResponseDTO();
            inventoryResponseDTO.setSellerId(queryRequest.getSellerId());
            inventoryResponseDTO.setSellerId(skuId);
            inventoryResponseDTO.setInventoryNum(inventoryNum);
            inventoryResponseDTOList.add(inventoryResponseDTO);

            inventoryBucketCache.threadLocalRemove();
        }
        return inventoryResponseDTOList;
    }


    /**
     * 检测扣减成功后的库存是否 触发回源(例如商品分桶库存1000，回源比例40%，那么实际剩余库存小于400就会触发回源库存的操作)
     *
     * @param bucketContext
     */
    private void checkInventoryBackSource(BucketContext bucketContext) {
        InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
        // 存储对应的需要校验扩容的分桶
        Map<String, Integer> bucketMap = bucketContext.getCapacityMap();
        // 获取当前的库存分桶配置
        InventoryBucketConfigDO inventoryBucketConfig = bucketContext.getInventoryBucketConfig();
        // 判断分桶当初分配的最大库存深度，计算是否触发回源比例
        List<BucketCacheBO> availableList = bucketContext.getAvailableList();
        if (CollUtil.isEmpty(availableList)) {
            return;
        }

        for (BucketCacheBO bucketCacheBO : availableList) {
            // 具体使用的哪个分桶扣减库存
            if (bucketMap.containsKey(bucketCacheBO.getBucketNo())) {
                // 这个是那个分桶扣完以后得库存数量
                Integer residueNum = bucketMap.get(bucketCacheBO.getBucketNo());
                // 当前分桶的分配总库存
                Integer bucketNum = bucketCacheBO.getBucketNum();
                // 触发回源比例的百分比, 回到中心桶去捞取一些库存回来放到这个分桶里去
                Integer backSourceProportion = inventoryBucketConfig.getBackSourceProportion();
                // 这里如果要更准确，需要用小数得到回源数，剩余数量小于回源数，那么就要回源
                // 这里省略了小数，所以可能会有一个数的误差，影响不大
                int backSourceNum = bucketNum * backSourceProportion / 100;
                // 回源比例的库存 大于 剩余的库存，触发异步扩容，或者没有返回剩余库存也说明扣减失败
                if (backSourceNum > residueNum) {   // 分桶剩余数量小于回源比例
                    // 标记出回源的具体分桶
                    inventoryDetail.setBucketNo(bucketCacheBO.getBucketNo());
                    sendAsynchronous(inventoryDetail);
                }
            }
        }
    }

    /**
     * 异步通知消息队列进行库存扩容
     *
     * @param inventoryDetail 库存扣减明细对象
     */
    private void sendAsynchronous(InventoryDetail inventoryDetail) {
        //1. 构建发送的消息对象
        BucketCapacity bucketCapacity = inventoryConverter.converter(inventoryDetail);
        //2. 发送消息，异步处理扩容
        bucketCapacityProducer.sendBucketCapacity(bucketCapacity);
    }

    /**
     * 将库存明细写入缓存(写入失败则代表已经被写入了，此时需要回退库存)
     *
     * @param bucketContext
     */
    private void writeInventoryDetail(BucketContext bucketContext) {
        // 明细记录
        InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();

        // 尝试写入明细记录，如果没有写入成功则说明库存需要回退
        Integer count = tairCache.exhsetNx(bucketContext.getInventoryDetailKey(), String.valueOf(inventoryDetail.getOrderId()), JSONObject.toJSONString(inventoryDetail));
        if (count < 0) {
            // 说明 明细已经存在了，写入失败 将库存回退到对应的分桶上
            tairCache.incr(inventoryDetail.getBucketNo(), inventoryDetail.getInventoryNum());
        } else {
            // 发送异步消息，写入库存扣减的明细入库
            inventoryDetailProducer.sendInventoryDetail(bucketContext.getInventoryDetail());
        }
    }

    /**
     * 扣减库存
     *
     * @param bucketContext 库存扣减上下文对象
     */
    private void deductInventory(BucketContext bucketContext) {
        List<BucketCacheBO> availableList = bucketContext.getAvailableList();
        InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
        Integer inventoryNum = inventoryDetail.getInventoryNum();
        Map<String, Integer> capacityMap = bucketContext.getCapacityMap();
        Boolean isDeduct = false;
        // 对分桶进行库存扣减(每次)
        for (BucketCacheBO bucketCacheBO : availableList) {
            String bucketNo = bucketCacheBO.getBucketNo();
            //自减，默认扣减后不能小于0，否则返回-1
            Integer residueNum = tairCache.decr(bucketNo, inventoryNum);
            // 库存扣减成功
            if (residueNum >= 0) {
                // 接下来 用于判断是否扩容的对象，只处理当前扣减成功的分桶的扩容消息发送。
                // 扣减失败的，在此时刻之前，肯定是已经有发送扩容消息的。
                // 如果消息没有消费到，或者消费失败，下次扣减库存的时候，也会再次发送消息。不需要重复发送了。
                capacityMap.put(bucketNo, residueNum);
                // 标记一下具体扣减的分桶属于哪个,以及标记当前剩余的库存
                inventoryDetail.setBucketNo(bucketNo);
                isDeduct = true;
                break;
            }
        }

        //分桶扣减都没有成功，此时抛出异常提示库存不足
        if (!isDeduct) {
            throw new BaseBizException(InventoryExceptionCode.INVENTORY_INSUFFICIENT_ERROR);
        }
    }

    /**
     * 验证当前得请求扣减是否已经存在了
     *
     * @param bucketContext
     */
    private String getRepeatDeductInfo(BucketContext bucketContext) {
        // 获取当前扣减库存对应订单的明细缓存
        return tairCache.exhget(bucketContext.getInventoryDetailKey(), String.valueOf(bucketContext.getInventoryDetail().getOrderId()));
    }

    /**
     * 构建接下来用于具体扣减库存所需要的模型对象
     *
     * @param request
     */
    private BucketContext buildDeductProductStock(InventoryRequest request) {
        // 1. 填充扣减库存相关信息的明细
        InventoryDetail inventoryDetail = inventoryConverter.converterRequest(request);
        // 2. 填充扣减库存的分桶配置信息
        BucketContext bucketContext = buildDeductBucketList(request);
        bucketContext.setInventoryDetail(inventoryDetail);
        return bucketContext;
    }

    /**
     * 填充扣减库存的分桶相关信息
     *
     * @param request
     * @return
     */
    private BucketContext buildDeductBucketList(InventoryRequest request) {
        BucketContext context = new BucketContext();
        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());

        // 获取本地缓存的分桶列表
        List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();

        if (CollUtil.isNotEmpty(availableList)) {
            // 获取本次访问请求的次数，用来定位具体使用的分桶库存
            Integer incrementCount = getIncrementCount(request);

            // 通过 运算得到本次访问所需要定位的分桶
            int index = incrementCount % availableList.size();
            // 获取本次准备处理的分桶信息,避免扣减失败(分桶已下线或者库存不足)，多备份几个

            BucketCacheBO bucketCacheBO = availableList.get(index);
            context.getAvailableList().add(bucketCacheBO);

            // 分桶数量大于3个才考虑重试，否则过少的分桶没有太大意义
            if (availableList.size() >= 3) {
                for (int i = 0; i < 2; i++) {
                    // 填充2个作为备份，如果超过下标则从0开始继续取
                    int num = index + i;
                    if (num >= availableList.size()) {
                        num = 0;
                    }
                    BucketCacheBO bucketCache = availableList.get(num);

                    context.getAvailableList().add(bucketCache);
                }
            }
        } else {
            //并发下，可能全部的分桶都下线了，这个时候使用中心桶进行库存扣减(因为其它分桶的下线库存回源会加到中心桶上)
            BucketCacheBO bucketCacheBO = new BucketCacheBO();
            bucketCacheBO.setBucketNo(buildSellerInventoryKey(request.getSellerId(), request.getSkuId()));
            //中心桶无需扩容，但是出现这种场景属于高并发下，分桶全部被下线了，此时需要保证分桶本地元数据和远程保持一致，为了性能，分桶下线未上粒度较大得锁
            // 所以需要当遇到使用中心桶的时候，再次触发一次远程缓存和本地缓存同步的操作，并且需要保证远程缓存最少有一个可用分桶存在
            bucketCacheBO.setBucketNum(0);
            context.getAvailableList().add(bucketCacheBO);
            // 异步消息发送同步本地缓存的消息
            bucketRefreshProducer.sendBucketOffline(request);
        }


        Long index = InventorBucketUtil.createDetailBucketKey(request.getOrderId(), bucketLocalCache.getInventoryBucketConfig().getBucketNum());

        String inventoryDetailKey = bucketLocalCache.getBucketDetailKeyList().get(Integer.valueOf(index + ""));

        context.setInventoryBucketConfig(bucketLocalCache.getInventoryBucketConfig());
        context.setInventoryDetailKey(inventoryDetailKey);
        inventoryBucketCache.threadLocalRemove();
        return context;
    }

    /**
     * 库存回退到中心桶上
     *
     * @param inventoryDetail
     */
    private void releaseInventory(InventoryDetail inventoryDetail) {
        // 对象为null，说明没有执行退货明细写入
        if (!Objects.isNull(inventoryDetail)) {
            String key = TairInventoryConstant.SELLER_INVENTORY_PREFIX + inventoryDetail.getSellerId() + inventoryDetail.getSkuId();
            // 将本次申请退回的库存，返回到中心桶上
            tairCache.incr(key, inventoryDetail.getInventoryNum());
        }
    }

    /**
     * 记录 退货库存记录
     *
     * @param request
     */
    private InventoryDetail saveRefundInventoryDetail(InventoryRequest request) {
        InventoryDetail inventoryDetail = inventoryRepository.getInventoryDetail(request.getOrderId());
        if (Objects.isNull(inventoryDetail)) {
            throw new BaseBizException(InventoryExceptionCode.INVENTORY_DETAIL_NULL_ERROR, InventoryExceptionCode.INVENTORY_DETAIL_NULL_ERROR.getErrorCode());
        }
        // 校验对应的退款库存申请记录
        List<InventoryDetail> inventoryDetails = inventoryRepository.queryRefundInventoryDetailList(request.getOrderId());
        Integer refundNum = 0;
        // 校验已退库存 + 本次申请退货库存 是否超过下单库存
        if (!CollectionUtils.isEmpty(inventoryDetails)) {
            for (InventoryDetail inventoryDetail1 : inventoryDetails) {
                refundNum = refundNum + inventoryDetail1.getInventoryNum();
            }
        }
        // 如果扣减的库存 大于已退库存和本次退的库存，则允许退货
        if (inventoryDetail.getInventoryNum() > (refundNum + request.getInventoryNum())) {
            InventoryDetail inventoryRefundDetail = inventoryConverter.converterRequest(request);
            inventoryRepository.saveInventoryDetail(inventoryRefundDetail, InventoryDeductionTypeEnum.INVENTORY_DETAIL_REFUND_TYPE.getCode());

            return inventoryRefundDetail;
        }

        return null;
    }

    /**
     * 验证查询入参
     *
     * @param queryRequest
     */
    private void checkQueryInventoryList(InventoryQueryRequest queryRequest) {
        if (Objects.isNull(queryRequest)) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
        if (Objects.isNull(queryRequest.getSellerId()) || CollectionUtils.isEmpty(queryRequest.getSkuIdList())) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
        if (queryRequest.getSkuIdList().size() > 100) {
            throw new BaseBizException(InventoryExceptionCode.PRODUCT_QUERY_CHECK_ERROR, InventoryExceptionCode.PRODUCT_QUERY_CHECK_ERROR.getErrorCode());
        }
    }

    /**
     * 验证扣减库存的入参是否非法
     *
     * @param request
     */
    private void checkDeductProductStock(InventoryRequest request) {
        if (Objects.isNull(request)) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
        if (Objects.isNull(request.getSellerId()) || Objects.isNull(request.getSkuId())
                || Objects.isNull(request.getInventoryNum()) || Objects.isNull(request.getOrderId())) {
            throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
        }
    }

    /**
     * 获取对应售卖商品的扣减访问次数
     * 这里考虑并发的时候自增导致性能过低，采取批量获取一批序号，当这批序号被使用完以后才会再次获取一次
     *
     * @param request
     * @return
     */
    private Integer getIncrementCount(InventoryRequest request) {
        String incrementKey = TairInventoryConstant.SELLER_SKU_STOCK_COUNT_PREFIX + request.getSellerId() + request.getSkuId();

        Long incrementCount = segmentNoGen.genNewNo(incrementKey);
        if (incrementCount > 0) {
            return incrementCount.intValue();
        }
        //避免获取缓存的时候出现异常，当为负数的时候默认取第一个，分桶最少存在1个
        return 0;
    }

    /**
     * 中心桶库存的key
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildSellerInventoryKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_INVENTORY_PREFIX + sellerId + skuId;
    }
}
